package com.example.tcp.server;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AsciiString;
import io.netty.util.AttributeKey;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;

@Data
@Slf4j
public abstract class Client {

    private static AttributeKey<Boolean> CLIENT_CREATED = AttributeKey.valueOf("CLIENT_CREATED");

    private static final Map<AsciiString, Client> clients = new ConcurrentHashMap<>();

    private static final ExecutorService taskService = Executors.newCachedThreadPool();

    protected AsciiString deviceId;
    protected Channel channel;

    private long lastRecvTime;
    private long recvTraffic;
    private long sendTraffic;
    private long loginTime = System.currentTimeMillis();
    protected AtomicInteger seqId = new AtomicInteger(1);
    private Map<Integer, PushPromise> pushPromiseMap = new ConcurrentHashMap<>();

    public static Optional<Client> getClient(AsciiString deviceId) {
        return Optional.ofNullable(clients.get(deviceId));
    }

    public static Client getOrCreateClient(AsciiString deviceId, Channel channel, Function<AsciiString, Client> creator) {
        return clients.compute(deviceId, new BiFunction<AsciiString, Client, Client>() {
            @Override
            public Client apply(AsciiString asciiString, Client client) {
                if(!channel.hasAttr(CLIENT_CREATED)) {
                    channel.attr(CLIENT_CREATED).set(true);
                    if(client != null) {
                        // 关掉前一个连接
                        log.info("close prev channel: {}, {}", deviceId, client.channel);
                        taskService.submit(() -> client.channel.close());
                    }
                    Client c = creator.apply(deviceId);
                    if(c == null) {
                        taskService.submit(() -> channel.close());
                    } else {
                        log.info("channel connect: {}, {}", deviceId, channel);
                    }
                    return c;
                }
                return client;
            }
        });

    }

    protected Client(AsciiString deviceId, Channel channel) {
        this.deviceId = deviceId;
        this.channel = channel;
        channel.closeFuture().addListener(f -> onDisconnect());
    }

    public ChannelFuture sendPushPromise(ByteBuf buf, long timeoutMs) {
        return sendPushPromise(buf, null, timeoutMs);
    }

    public ChannelFuture sendPushPromise(Object msg, Object obj, long timeoutMs) {
        ChannelFuture future =  send(msg);
        future.addListener(f -> {
            pushPromiseMap.put(seqId.incrementAndGet(), new PushPromise(obj, timeoutMs));
        });
        return future;
    }

    public ChannelFuture send(Object msg) {
        ChannelFuture future = channel.writeAndFlush(msg);
        return future;
    }

    public Optional<PushPromise> onRecvPushAck(int seqId) {
        return Optional.ofNullable(pushPromiseMap.remove(seqId));
    }

    public void onRecv(Frame frame) {
        updateRecvTraffic(frame.content().readableBytes());
        lastRecvTime = System.currentTimeMillis();
        onRecvFrame(frame);
    }

    protected abstract void onRecvFrame(Frame frame);

    public void updateSendTraffic(long size) {
        sendTraffic += size;
    }

    private void updateRecvTraffic(long size) {
        recvTraffic += size;
    }

    private PushPromise removePushPromise(int seqId) {
        return pushPromiseMap.remove(seqId);
    }

    private void onDisconnect() {
        this.onClientDisconnect();
        this.pushPromiseMap.forEach(new BiConsumer<Integer, PushPromise>() {
            @Override
            public void accept(Integer integer, PushPromise promise) {
                promise.getFuture().complete(0);
            }
        });
        clients.compute(this.deviceId, new BiFunction<AsciiString, Client, Client>() {
            @Override
            public Client apply(AsciiString asciiString, Client old) {
                if(old != null && old.channel.equals(channel)) {
                    return null;
                }
                return old;
            }
        });
    }

    public Integer incrementAndGetSeqId() {
        return seqId.getAndIncrement();
    }

    public abstract void onClientDisconnect();


    public static void onChannelActive(ChannelHandlerContext ctx) {
        log.info("channel active: {}", ctx.channel());
    }

    public static void onChannelInactive(ChannelHandlerContext ctx) {
        log.info("channel inactive: {}", ctx.channel());
    }

}
